package com.telra.app.server.esper;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.Topic;

import com.telra.app.server.data.OrderAggregate;
import com.telra.app.server.data.OrderEvent;
import com.telra.app.server.protobuff.AggProtos.AggGUIUpdate;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.impl.Log4JLogger;


/**
 * Processes CEP updates from Esper and publishes to desktop
 *
 * @author jamesarlet
 *
 */
public class AggregateListener {

    private static final Log4JLogger log = (Log4JLogger) LogFactory.getFactory().getInstance(AggregateListener.class);

    private Map<OrderEvent.Category, OrderAggregate> map;

    public AggregateListener() {
        map = new ConcurrentHashMap<OrderEvent.Category, OrderAggregate>();
        map.put(OrderEvent.Category.ACCESSORIES, new OrderAggregate(OrderEvent.Category.ACCESSORIES,0,0.0,0.0,0.0));
        map.put(OrderEvent.Category.CLOTHING, new OrderAggregate(OrderEvent.Category.CLOTHING,0,0.0,0.0,0.0));
        map.put(OrderEvent.Category.FOOTWEAR, new OrderAggregate(OrderEvent.Category.FOOTWEAR,0,0.0,0.0,0.0));
        map.put(OrderEvent.Category.MISC, new OrderAggregate(OrderEvent.Category.MISC,0,0.0,0.0,0.0));
        map.put(OrderEvent.Category.OUTERWEAR, new OrderAggregate(OrderEvent.Category.OUTERWEAR,0,0.0,0.0,0.0));

        Thread t = new Thread(new PublishUpdate());
        t.setName("publish update");
        t.start();
    }


    public void update(OrderEvent.Category category, int count, double avg, double sum, double max) {
        OrderAggregate agg = map.get(category);
        agg.setAvg(avg);
        agg.setSum(sum);
        agg.setMax(max);
        agg.setCount(count);
    }

    private class PublishUpdate implements Runnable {

        public void run() {

            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
            connectionFactory.setUseAsyncSend(true);

            Connection connection;
            try {
                connection = connectionFactory.createConnection();

                connection.start();

                Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

                Topic topic = session.createTopic("AGG.OUT");
                MessageProducer producer = session.createProducer(topic);
                producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
                log.info("started agg publisher to gui");

                while (true) {	//send to the gui every 2 seconds
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                    //publish protobuff over MQ
                    ObjectMessage bMessage = session.createObjectMessage();

                    for (OrderAggregate order : map.values()) {
                        //convert to protobuff cat

                        OrderEvent.Category cat = order.getCategory();

                        AggGUIUpdate.Category aggCat =
                                (cat == OrderEvent.Category.ACCESSORIES ? AggGUIUpdate.Category.ACCESSORIES:
                                        cat == OrderEvent.Category.CLOTHING ? AggGUIUpdate.Category.CLOTHING:
                                                cat == OrderEvent.Category.FOOTWEAR ? AggGUIUpdate.Category.FOOTWEAR:
                                                        cat == OrderEvent.Category.MISC ? AggGUIUpdate.Category.MISC:
                                                                AggGUIUpdate.Category.OUTERWEAR
                                );


                        AggGUIUpdate o = AggGUIUpdate.newBuilder()
                                .setItemCategory(aggCat)
                                .setCount(order.getCount())
                                .setAvg(order.getAvg())
                                .setSum(order.getSum())
                                .setMax(order.getMax())
                                .build();

                        bMessage.setObject(o);
                        producer.send(bMessage);
                        log.info("published to gui: " + order);

                    }

                }
            }
            catch (JMSException e1) {
                e1.printStackTrace();
            }
        }

    }

}
